package com.lixinlei.jedis.demo.delayqueue;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import redis.clients.jedis.Jedis;

import java.lang.reflect.Type;
import java.util.Set;
import java.util.UUID;

/**
 * 基于 Redis 的 zset 实现延迟队列
 * - 元素的过期时间作为 score；
 * - 每次从集合中取的时候，只取到期的元素；
 * @param <T>
 */
public class RedisDelayQueue<T> {

    static class TaskItem<T> {
        public String id;
        public T msg;
    }

    // JSON.parseObject(s, TaskType) 在把 Json 字符串转换成对象的时候，需要指定一个 Type 类型的参数，
    // 如果对象的类本身有泛型，就需要被这个类包装在 TypeReference 中；
    private Type TaskType = new TypeReference<TaskItem<T>>() {}.getType();

    private Jedis jedis;
    private String queueKey;

    public RedisDelayQueue(Jedis jedis, String queueKey) {
        this.jedis = jedis;
        this.queueKey = queueKey;
    }

    public void delay(T msg) {
        TaskItem<T> task = new TaskItem<>();
        task.id = UUID.randomUUID().toString();
        task.msg = msg;

        String s = JSON.toJSONString(task);

        // queueKey - key（集合名）；
        // System.currentTimeMillis() + 5000 - score，把元素的过期时间作为 score；
        // s - member（集合中的元素）；
        jedis.zadd(queueKey, System.currentTimeMillis() + 5000, s);
    }

    public void loop() {
        while (!Thread.interrupted()) {
            // 按 score 的范围从集合中获取元素，没到期的元素就取不出来；
            Set<String> values = jedis.zrangeByScore(queueKey, 0, System.currentTimeMillis(), 0, 1);
            if (values.isEmpty()) {
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    break;
                }
                continue;
            }
            // 到期的元素就可以取出来
            String s = values.iterator().next();
            if (jedis.zrem(queueKey, s) > 0) {
                TaskItem<T> task = JSON.parseObject(s, TaskType);
                this.handleMst(task.msg);
            }
        }
    }

    public void handleMst(T msg) {
        System.out.println(msg);
    }

    public static void main(String[] args) {
        Jedis jedis = new Jedis();
        RedisDelayQueue<String> queue = new RedisDelayQueue<>(jedis, "q-demo");

        Thread producer = new Thread() {
            @Override
            public void run() {
                for (int i = 0; i < 10; i++) {
                    queue.delay("delay-set" + i);
                }
            }
        };

        Thread consumer = new Thread() {
            @Override
            public void run() {
                queue.loop();
            }
        };

        producer.start();
        consumer.start();

        try {
            // 让 Main 线程等 producer 执行完
            producer.join();
            // producer 线程执行完了，Main 线程再睡 6s
            Thread.sleep(6000);
            // 给 consumer 发一个 interrupt 信号
            consumer.interrupt();
            // 让 Main 线程等 consumer 执行完
            consumer.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}
